/*
 * Copyright 2012 The Netty Project
 *
 * The Netty Project licenses this file to you under the Apache License,
 * version 2.0 (the "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at:
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */
package com.qxp.simplerpc.connect;

import com.qxp.simplerpc.entity.InvokerMessage;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.EventLoop;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * Handler implementation for the object echo com.qxp.example.  It initiates the
 * ping-pong traffic between the object echo com.qxp.example and com.qxp.example by sending the
 * first message to the com.qxp.example.
 */
@Component
@ChannelHandler.Sharable
public class RpcClientHandler extends ChannelInboundHandlerAdapter {
    private static Logger logger = LoggerFactory.getLogger(RpcClientHandler.class);
    // 用于接收客户端返回结果
    private Map<String, Object> resultMap = new ConcurrentHashMap<>();
    // 通道句柄内容
    private ChannelHandlerContext ctx;
    // 同步的方法
    private CountDownLatch countDownLatch = new CountDownLatch(1);
    //
    @Autowired
    private RpcClient rpcClient;

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        setCtx(ctx);
        this.countDownLatch.countDown();
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        InvokerMessage invokerMessage = (InvokerMessage) msg;
        Object result = invokerMessage.getResult();
        String token = invokerMessage.getToken();
        resultMap.put(token, result);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }

    /**
     * retry to connect server when exception caught
     * @param ctx
     * @param cause
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        logger.error("socket exception,caused by:", cause);
        final EventLoop eventExecutors = ctx.channel().eventLoop();
        eventExecutors.schedule(rpcClient, 1L, TimeUnit.SECONDS);
    }
    /**
     * 向客户端发送执行的消息体
     *
     * @param msg 消息对象
     */
    public void send(Object msg) {
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        getCtx().writeAndFlush(msg);
    }

    /**
     * 获取server中返回的结果
     *
     * @param token
     * @return
     * @throws InterruptedException
     */
    public Object getMethodInvokerResult(String token) throws InterruptedException {
        while (Objects.isNull(resultMap.get(token))) {
            Thread.sleep(50);
        }
        Object result = resultMap.get(token);
        resultMap.remove(token);
        return result;
    }

    public ChannelHandlerContext getCtx() {
        return ctx;
    }

    public void setCtx(ChannelHandlerContext ctx) {
        this.ctx = ctx;
    }

    public RpcClient getRpcClient() {
        return rpcClient;
    }

    public void setRpcClient(RpcClient rpcClient) {
        this.rpcClient = rpcClient;
    }
}
